package cn._51doit.day06;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.HashMap;
import java.util.Map;

/**
 * @create: 2021-10-22 00:25
 * @author: 今晚打脑斧先森
 * @program: MyKeyedStateDemo
 * @Description:
 *
 *   自己写一个Flink的程序，将中间结果计算出来，不使用Flink的状态API
 *   解决各自累加各自Key的次数
 *   统计key的总数
 *
 *   使用map,让每个分区内的单词按照自己的个数统计,如果不用map的话,那么每个分区会公用一个中间变量
 **/
public class MyKeyedStateDemo {
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port", 8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(configuration);

        //开启checkpoint,默认为无限重启
        env.enableCheckpointing(10000);

        //spark,2
        DataStreamSource<String> lines = env.socketTextStream("doit01", 8888);

        SingleOutputStreamOperator<Tuple2<String, Integer>> tpStream = lines.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                String[] split = value.split(",");
                String word = split[0];
                if (word.startsWith("error")) {
                    throw new RuntimeException("数据出问题了");
                }
                int count = Integer.parseInt(split[1]);
                return Tuple2.of(word, count);
            }
        });
        //sum底层调用的是reduce方法，再StreamGroupedReduceOperator的processElement使用和更新了状态
        KeyedStream<Tuple2<String, Integer>, String> keyedStream = tpStream.keyBy(t -> t.f0);
        SingleOutputStreamOperator<Tuple2<String, Integer>> res = keyedStream.map(new MyReduceFunction());
        res.print();
        env.execute();


    }
    private static class MyReduceFunction extends RichMapFunction<Tuple2<String,Integer>,Tuple2<String,Integer>>{

        private Map<String,Integer> wc = new HashMap<>();
        @Override
        public Tuple2<String, Integer> map(Tuple2<String, Integer> in) throws Exception {
            String word = in.f0;
            Integer currentCount = in.f1; //当前的个数
            Integer historyCount = wc.get(word); //历史的个数
            if (historyCount==null){
                historyCount=0;
            }
            int totalCount = currentCount+historyCount;//总的个数
            //将数据更新到map中
            wc.put(word,totalCount);
            //输出数据
            in.f1=totalCount;
            return in;
        }
    }
}
